package com.xpc.rpc.consumer;

import com.xpc.proxy.common.consumer.Consumer;
import com.xpc.rpc.codec.RpcDecoder;
import com.xpc.rpc.codec.RpcEncoder;
import com.xpc.rpc.consumer.cache.RpcConsumerConfigCache;
import com.xpc.rpc.consumer.handler.RpcConsumerHandler;
import com.xpc.rpc.protocol.ProtocolMessage;
import com.xpc.rpc.protocol.meta.ServiceMeta;
import com.xpc.rpc.protocol.request.RpcRequest;
import com.xpc.rpc.register.common.RegisterService;
import com.xpc.rpc.register.common.config.RegisterConfig;
import com.xpc.rpc.register.zookeeper.ZookeeperRegisterServiceImpl;
import io.netty.bootstrap.Bootstrap;
import io.netty.channel.*;
import io.netty.channel.nio.NioEventLoopGroup;
import io.netty.channel.socket.SocketChannel;
import io.netty.channel.socket.nio.NioSocketChannel;
import io.netty.util.concurrent.Future;
import io.netty.util.concurrent.GenericFutureListener;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.beans.BeanUtils;

import java.util.Map;
import java.util.concurrent.ConcurrentHashMap;

public class RpcConsumer implements Consumer {

     private static final Logger LOGGER = LoggerFactory.getLogger(RpcConsumer.class);

     private Bootstrap bootstrap;

     private EventLoopGroup eventLoopGroup;

    private static volatile RpcConsumer instance;

    private Map<String,RpcConsumerHandler> handlerMap = new ConcurrentHashMap<>();

     private RegisterService registerService;

     public RpcConsumer() {
         eventLoopGroup = new NioEventLoopGroup();
         bootstrap = new Bootstrap();

         bootstrap.group(eventLoopGroup)
                 .channel(NioSocketChannel.class)
                 .handler(new ChannelInitializer<SocketChannel>() {
                     @Override
                     protected void initChannel(SocketChannel channel) throws Exception {
                         ChannelPipeline pipeline = channel.pipeline();
                         //我们自己的编解码器
                         pipeline.addLast(new RpcDecoder());
                         pipeline.addLast(new RpcEncoder());
                         pipeline.addLast(new RpcConsumerHandler());
                     }
                 });
     }

    public static RpcConsumer getInstance(){
        if (instance == null){
            synchronized (RpcConsumer.class){
                if (instance == null){
                    instance = new RpcConsumer();
                }
            }
        }
        return instance;
    }



    @Override
     public Object sendRequest(ProtocolMessage<RpcRequest> requestProtocolMessage) {
        this.registerService = new ZookeeperRegisterServiceImpl();
        RegisterConfig registerConfig = new RegisterConfig();
        BeanUtils.copyProperties(RpcConsumerConfigCache.getCache(),registerConfig);
        this.registerService.init(registerConfig);

         ServiceMeta serviceMeta = null;
         try {
              serviceMeta = registerService.discovery(requestProtocolMessage.getT().getClassName(),registerConfig.getLoadbalancerType());
         } catch (Exception e) {
             e.printStackTrace();
         }
         RpcRequest request = requestProtocolMessage.getT();
         String key = buildHandlerMapKey(request.getClassName(), request.getMethodName());
         RpcConsumerHandler consumerHandler;
         if(handlerMap.containsKey(key)) {
             consumerHandler = handlerMap.get(key);
             Channel channel = consumerHandler.getChannel();
             if(!channel.isOpen() || !channel.isActive()) {
                 consumerHandler = getConsumerHandler(key, serviceMeta.getServerAddress(), serviceMeta.getRegisterPort());
                 handlerMap.put(buildHandlerMapKey(request.getClassName(),request.getMethodName()),consumerHandler);
             }
         }else {
             consumerHandler = getConsumerHandler(key, serviceMeta.getServerAddress(), serviceMeta.getRegisterPort());
             if(consumerHandler == null) {
                 throw new RuntimeException("");
             }
             handlerMap.put(buildHandlerMapKey(request.getClassName(),request.getMethodName()),consumerHandler);
         }
         return consumerHandler.sendRequest(requestProtocolMessage);
     }

     private RpcConsumerHandler getConsumerHandler(String key,String host, int port) {
        try {
            ChannelFuture channelFuture = bootstrap.connect(host, port).sync();
            channelFuture.addListener(new GenericFutureListener<Future<? super Void>>() {
                @Override
                public void operationComplete(Future<? super Void> future) throws Exception {
                    if(future.isSuccess()) {
                        LOGGER.info("客户端连接成功........");
                    }else {
                        LOGGER.info("客户端连接失败........");
                    }
                }
            });
            RpcConsumerHandler rpcConsumerHandler = channelFuture.channel().pipeline().get(RpcConsumerHandler.class);
            handlerMap.put(key,rpcConsumerHandler);
            return rpcConsumerHandler;
        } catch (InterruptedException e) {
            return null;
        }
    }


     public void close() {
         if(eventLoopGroup != null) {
             eventLoopGroup.shutdownGracefully();
         }
     }

     private String buildHandlerMapKey(String className,String methodName) {
         StringBuilder key = new StringBuilder();
         key.append(className).append("#").append(methodName);
         return key.toString();
     }

}
